package com.pay.buffer;

import com.alibaba.fastjson.JSONObject;
import com.pay.config.RabbitMQBatchConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author zhouyihe
 * @ClassName MqBatchDeliveryBuffer
 * @qq 1552951165
 */
@Component
public class MqBatchDeliveryBuffer<E> {
    private static final Logger log = LoggerFactory.getLogger(MqBatchDeliveryBuffer.class);
    /**
     * 定义容器存放msg
     */
    private LinkedBlockingQueue<E> buffers = new LinkedBlockingQueue<E>();
    private volatile boolean isRun = true;
    
    
    @Autowired
    private AmqpTemplate amqpTemplate;
    
    public MqBatchDeliveryBuffer() {
        // new TaskBufferThread().start();
        // 利用单例线程池启动该线程
        Executors.newSingleThreadExecutor().execute(
                new TaskBufferThread()
        );
    }
    
    
    /**
     * 将msg消息暂时存放在我们缓冲区中
     *
     * @param e
     */
    public void add(E e) {
        buffers.offer(e);
    }
    
    
    /**
     * 合并消息线程（将多条消息合并成一条msg ，投递MQ中）
     */
    class TaskBufferThread implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    // 合并多条消息
                    List<E> list = new ArrayList<>();
                    for (Integer i = 0; i < RabbitMQBatchConfig.BUFFER_SIZE; i++) {
                        // 超时时间是为1s 1s 没有取到情况下 则 返回null
                        E e = buffers.poll(RabbitMQBatchConfig.BUFFER_INTERVAL_TIME, TimeUnit.SECONDS);
                        if (e != null) list.add(e);
                    }
                    // 批量的形式投递到MQ中
                    String msg = JSONObject.toJSONString(list);
                    amqpTemplate.convertAndSend(RabbitMQBatchConfig.EXCHANGE_BATCH_NAME,"",msg);
                } catch (InterruptedException e) {
                    // throw new RuntimeException(e);
                    log.error("e:{}",e);
                }
            }
        }
    }
    
    
    // 不合并消息,直接发送到mq
    // class TaskBufferThread implements Runnable{
    //     @Override
    //     public void run() {
    //         while (true){
    //             try {
    //                 buffers.take();
    //             } catch (InterruptedException e) {
    //                 //throw new RuntimeException(e);
    //                 e.printStackTrace();
    //             }
    //         }
    //     }
    // }
    
    // for test
    // public static void main(String[] args) throws InterruptedException {
    //     LinkedBlockingQueue<String> strings = new LinkedBlockingQueue<>();
    //     strings.offer("1");
    //     strings.offer("2");
    //     strings.offer("3");
    //     System.out.println(strings.take());
    //     System.out.println(strings.take());
    //     System.out.println(strings.take());
    //     System.out.println(strings.take());
    // }
}
